package com.zyx.flinkdemo.stream.param;

import com.zyx.flinkdemo.stream.utils.KafkaUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * @author zyx
 * @since 2021/5/24 20:55
 * desc: 参数传递实例
 */
@Slf4j
public class ParamTransDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        ParameterTool params = ParameterTool.fromArgs(args);
        String runEnv = params.get("runEnv");
        String kafkaServer = KafkaUtil.getKafkaServer(runEnv);

        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.12.13.22:9092,10.12.13.23:9092,10.12.13.24:9092");
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "wc_consumer_05251405");
        prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("kafka_topic_wc", new SimpleStringSchema(), prop);
        kafkaConsumer.setStartFromGroupOffsets();
        DataStreamSource<String> kafkaDS = env.addSource(kafkaConsumer);

        // KafkaFLinkConsumer
        /*DataStreamSource<String> kafkaDS = env.addSource(
                KafkaUtil.getKafkaSource("kafka_topic_wc", "wc_consumer_05251341", kafkaServer));*/
        DataStream<Tuple2<String, Long>> counts = kafkaDS
                .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(new Tuple2<>(word, 1L));
                        }
                    }
                })
                .keyBy((KeySelector<Tuple2<String, Long>, Object>) value -> value.f0)
                .sum(1);

        counts.print();
        // 执行环境
        env.execute();
    }
}
